package com.ict.utils;


import com.google.protobuf.InvalidProtocolBufferException;
import com.ict.config.Config;
import com.ict.dataprocess.SyncFlagClass;

//import Database_Storage.RedisStorage;

import com.rabbitmq.client.*;

import team.iOceanPlus.PB.Target;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeoutException;

public class MQTopicConsume {

    // 可以方便地修改日期格式
    private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");

    private static String m_IP,m_user,m_password,m_EXCHANGE_NAME;
    private static int m_port;
    private static String EXCHANGE_TYPE = "topic";
    private static boolean EXCHANGE_DURABLE = true;

    private static String m_QUEUE_NAME;
    private static int QUEUE_NAME_LENGTH = 15;
    private static boolean QUEUE_DURABLE = false;
    private static boolean QUEUE_EXCLUSIVE = true;
    private static boolean QUEUE_AUTO_DELETE = false;
    private static String[] m_routingKeys;
    private static int message_ttl = 300000;    // 单位 ms

    private static Connection connection;
    private static Channel channel;

    //RedisStorage redisStorage;
    //MysqlStorage mysqlStorage;

    // 各路routingKey
    private static String Orig_routingKey = "OnLine.PreprocessedData.#";
    private static String Fusion_routingKey = "OnLine.FusedData.TargetTracks";
    private static String Anomaly_routingKey = "OnLine.Alert.Anomaly";
    private static String ZoneEntry_routingKey = "OnLine.Alert.ZoneEntry";
    private static String DeleteTarget_routingKey = "Command.DeleteTarget";
    private static String UpdateTarget_routingKey = "Command.UpdateTargetProperty";

    private static String Monitor_routingKey = "Monitor.Probe";

    private static int preprocessRcvd = 0,
                       fusedRcvd = 0,
                       anomalyRcvd = 0,
                       commandRcvd = 0;

    public MQTopicConsume(String IP, int port, String user, String password,
                          String EXCHANGE_NAME, String[] routingKeys, String queueName_pre) {

        m_IP= IP;
        m_port = port;
        m_user = user;
        m_password = password;
        m_EXCHANGE_NAME = EXCHANGE_NAME;
        m_routingKeys = routingKeys;
        m_QUEUE_NAME = queueName_pre;

        BaseConnection();

        //redisStorage = new RedisStorage();
        //mysqlStorage = new MysqlStorage();
    }

    protected void finalize() throws Throwable {

        super.finalize();

        channel.close();
        connection.close();
    }

    /**
     * 创建连接连接到RabbitMQ
     * @return   0 means success
     *           -1 means failure
     */
    private int BaseConnection(){

        ConnectionFactory factory = new ConnectionFactory();
        //设置MabbitMQ所在主机ip或者主机名
        factory.setHost(m_IP);
        factory.setPort(m_port);
        factory.setUsername(m_user);
        factory.setPassword(m_password);
        // 自动重连
        factory.setAutomaticRecoveryEnabled(true);

        //创建一个连接
        try {
            connection = factory.newConnection();
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ 请确认输入的IP地址、用户名、密码是否准确！");
            return  -1;
        } catch (TimeoutException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]连接RabbitMQ超时，重试中......");
            BaseConnection();
            return  -1;
        }

        //创建一个频道
        try {
            channel = connection.createChannel();
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ 创建频道出错，重试中......");
            BaseConnection();
            return -1;
        }

        //声明一个路由器，指定名称、模式、及其是否durable
        try {
            channel.exchangeDeclare(m_EXCHANGE_NAME,EXCHANGE_TYPE,EXCHANGE_DURABLE);
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ 路由器声明失败，重试中......");
            BaseConnection();
            return -1;
        }

        //指定一个队列,随机队列名,non-durable,exclusive,not auto-delete
        m_QUEUE_NAME += getRandomString(QUEUE_NAME_LENGTH);

        Map<String,Object> args = new HashMap();
        args.put("x-message-ttl",message_ttl);
        try {
            channel.queueDeclare(m_QUEUE_NAME,QUEUE_DURABLE,QUEUE_EXCLUSIVE,QUEUE_AUTO_DELETE,args);
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ 队列声明失败，重试中......");
            System.out.println(e.getMessage());
            //BaseConnection();
        }

        //指定路由关键字
        for(String bindingKey:m_routingKeys){
            try {
                channel.queueBind(m_QUEUE_NAME,m_EXCHANGE_NAME,bindingKey);
            } catch (IOException e) {

                System.out.println("["+ dateFormat.format( new Date() ) +"]MQ 队列绑定路由关键字 " + bindingKey + " 时出错!重试中......");
                BaseConnection();
            }
        }

        System.out.println("["+ dateFormat.format( new Date() ) +"]MQ 订阅消息者成功连接至RabbitMQ !");

        return 0;

    }

    /**
     * 订阅消息函数
     */
    public void ConsumeOrig(){

        System.out.println("["+ dateFormat.format( new Date() ) +"]MQ Receiver " + m_QUEUE_NAME + " Waiting for messages.To exit press CTRL+C.");

        //消息订阅
        Consumer consumer = new DefaultConsumer(channel){

            /*
            * 处理接收到的消息
            * */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            	handleReceivedOrig(consumerTag, envelope, properties, body);

            }

        };

        // 消息反馈,消息消费完成确认
        try {
            channel.basicConsume(m_QUEUE_NAME,true,consumer);
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ Receiver "+ m_QUEUE_NAME + " failed to basic consume." );
        }

    }
    public void ConsumeFuse(){

        System.out.println("["+ dateFormat.format( new Date() ) +"]MQ Receiver " + m_QUEUE_NAME + " Waiting for messages.To exit press CTRL+C.");

        //消息订阅
        Consumer consumer = new DefaultConsumer(channel){

            /*
            * 处理接收到的消息
            * */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            	handleReceivedFuse(consumerTag, envelope, properties, body);
            	
            		

            }

        };

        // 消息反馈,消息消费完成确认
        try {
            channel.basicConsume(m_QUEUE_NAME,true,consumer);
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ Receiver "+ m_QUEUE_NAME + " failed to basic consume." );
        }

    }

    private void handleReceivedOrig(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {

    		
			
			UUID  uuid=UUID.randomUUID();
			
			List<byte []>  lstCache=CacheObject.returnLstCache();
			if(lstCache.size()==Config.listCatchSize) {
				
			}else {
				lstCache.add(body);
			}
			//if(ConnRedisUtil.jedis.exists("recv_orig_pb")) {
				
				/*if(ConnRedisUtil.jedis.hkeys("recv_orig_pb").size()==3) {
					
				}else {
					ConnRedisUtil.jedis.hset("recv_orig_pb".getBytes(), (uuid.toString()).getBytes(), body);
				}
				*/
				
			//}
			/*if((!ConnRedisUtil.jedis.exists("recv_orig_pb"))&&(!ConnRedisUtil.jedis.exists("cp_recv_orig_pb"))) {
				ConnRedisUtil.jedis.hset("recv_orig_pb".getBytes(), (uuid.toString()).getBytes(), body);
			}
			
*/		/*	
			try {
				Target.PBTarget test= Target.PBTarget.parseFrom(body);
			} catch (InvalidProtocolBufferException e) {
				e.printStackTrace();
			}
			*/
		
		
    }
    private void handleReceivedFuse(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {

		
		
		UUID  uuid=UUID.randomUUID();
		
		List<byte []>  lstCache=CacheObject.returnLstCache();
		if(lstCache.size()==Config.listCatchSize) {
			
		}else {
			lstCache.add(body);
		}
		//if(ConnRedisUtil.jedis.exists("recv_orig_pb")) {
			
			/*if(ConnRedisUtil.jedis.hkeys("recv_orig_pb").size()==3) {
				
			}else {
				ConnRedisUtil.jedis.hset("recv_orig_pb".getBytes(), (uuid.toString()).getBytes(), body);
			}
			*/
			
		//}
		/*if((!ConnRedisUtil.jedis.exists("recv_orig_pb"))&&(!ConnRedisUtil.jedis.exists("cp_recv_orig_pb"))) {
			ConnRedisUtil.jedis.hset("recv_orig_pb".getBytes(), (uuid.toString()).getBytes(), body);
		}
		
*/		/*	
		try {
			Target.PBTarget test= Target.PBTarget.parseFrom(body);
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
		*/
	
	
}
    
    
 private void handleReceivedDel(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {

		
		
	    List<MessageStruct> lstMsg = new ArrayList<MessageStruct>();
		MessageStruct msgStruct = new MessageStruct(Config.forwardRoutingKeys[0],body);
		lstMsg.add(msgStruct);
		//connMQUtil.sendClose();
		PublishInit.publish.Publish(lstMsg);
		//if(ConnRedisUtil.jedis.exists("recv_orig_pb")) {
			
			/*if(ConnRedisUtil.jedis.hkeys("recv_orig_pb").size()==3) {
				
			}else {
				ConnRedisUtil.jedis.hset("recv_orig_pb".getBytes(), (uuid.toString()).getBytes(), body);
			}
			*/
			
		//}
		/*if((!ConnRedisUtil.jedis.exists("recv_orig_pb"))&&(!ConnRedisUtil.jedis.exists("cp_recv_orig_pb"))) {
			ConnRedisUtil.jedis.hset("recv_orig_pb".getBytes(), (uuid.toString()).getBytes(), body);
		}
		
*/		/*	
		try {
			Target.PBTarget test= Target.PBTarget.parseFrom(body);
		} catch (InvalidProtocolBufferException e) {
			e.printStackTrace();
		}
		*/
	
	
}
    private void  forWard(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {

		
		
 			
 			ConnMQUtil connMQUtil = new ConnMQUtil();
 			try {
				connMQUtil.send("topic", Config.sendExChangeName,  m_routingKeys[0], null, body);
				/*MQTopicPublish publish=new  MQTopicPublish();
				publish.Publish(messages);*/
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
 		
 		
     }
    
    
    public void ConsumeFoward(){

        System.out.println("["+ dateFormat.format( new Date() ) +"]MQ Receiver " + m_QUEUE_NAME + " Waiting for messages.To exit press CTRL+C.");

        //消息订阅
        Consumer consumer = new DefaultConsumer(channel){

            /*
            * 处理接收到的消息
            * */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

            	forWard(consumerTag, envelope, properties, body);

            }

        };

        // 消息反馈,消息消费完成确认
        try {
            channel.basicConsume(m_QUEUE_NAME,true,consumer);
        } catch (IOException e) {

            System.out.println("["+ dateFormat.format( new Date() ) +"]MQ Receiver "+ m_QUEUE_NAME + " failed to basic consume." );
        }

    }
    

    /**
     * 用于生成随机队列名
     * @param length  随机队列名长度
     * @return  随机队列名
     */
    private static String getRandomString(int length){
        String str = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";//含有字符和数字的字符串
        Random random = new Random();//随机类初始化
        StringBuffer sb = new StringBuffer();//StringBuffer类生成，为了拼接字符串

        for (int i = 0; i < length; ++i) {
            int number = random.nextInt(62);// [0,62)

            sb.append(str.charAt(number));
        }
        return sb.toString();
    }

}
